/**
 * 
 */
package edu.umd.clip.lm.storage;

import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.util.concurrent.locks.*;
import java.util.concurrent.*;
import java.util.*;

import edu.umd.clip.lm.model.OnDiskCompactProbTree;
import edu.umd.clip.lm.model.ProbTree;
import edu.umd.clip.lm.util.*;

/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 */
public class CachedProbTreeStorage extends AbstractProbTreeStorage {
	private AbstractProbTreeStorage backend;
	public static final int DEFAULT_HARD_CACHE_SIZE = 5000;
	public static final int DEFAULT_SOFT_CACHE_SIZE = 15000;
	private static final int NUM_CACHES = 16;
	
	private Cache caches[];
	
	// use a wrapper to allow the cache to hold negative results (null values)
	private static class RequestResult {
		OnDiskCompactProbTree tree;
		private RequestResult(OnDiskCompactProbTree tree) {
			this.tree = tree;
		}
	}
	
	private class Cache extends LRUCache<Key,RequestResult> implements AsyncConsumer<Key,OnDiskCompactProbTree>, AsyncProducer<Key,OnDiskCompactProbTree> {
		private AsyncConsumerMultiplexor<Key,OnDiskCompactProbTree> multiplexor;
		
		public Cache(int hardCapacity, int softCapacity) {
			super(hardCapacity, softCapacity);
			multiplexor = new AsyncConsumerMultiplexor<Key,OnDiskCompactProbTree>(backend) {

				@Override
				public void receivedData(Key key, OnDiskCompactProbTree value) {
					putItem(key, new RequestResult(value));
					super.receivedData(key, value);
				}
			};
		}

		/* (non-Javadoc)
		 * @see edu.umd.clip.lm.util.LRUCache#loadItem(java.lang.Object)
		 */
		@Override
		protected RequestResult loadItem(Key key) {
			OnDiskCompactProbTree tree = backend.getProbTree(key);
			return new RequestResult(tree);
		}

		/* (non-Javadoc)
		 * @see edu.umd.clip.lm.storage.AsyncConsumer#receivedData(java.lang.Object, java.lang.Object)
		 */
		@Override
		public void receivedData(Key key, OnDiskCompactProbTree value) {
			multiplexor.receivedData(key, value);
		}

		/* (non-Javadoc)
		 * @see edu.umd.clip.lm.storage.AsyncProducer#request(java.lang.Object, edu.umd.clip.lm.storage.AsyncConsumer)
		 */
		@Override
		public void request(Key key, AsyncConsumer<Key, OnDiskCompactProbTree> callback) {
			RequestResult result = peekItem(key);
			if (result != null) {
				callback.receivedData(key, result.tree);
			} else {
				multiplexor.request(key, callback);
			}
		}
	}
	
	public CachedProbTreeStorage(AbstractProbTreeStorage backend, int hardCacheSize, int softCacheSize) {
		this.backend = backend;
		this.caches = new Cache[NUM_CACHES];
		for(int i=0; i<NUM_CACHES; ++i) {
			caches[i] = new Cache(hardCacheSize/NUM_CACHES, softCacheSize/NUM_CACHES);
		}
	}

	public CachedProbTreeStorage(AbstractProbTreeStorage backend) {
		this(backend, DEFAULT_HARD_CACHE_SIZE,	DEFAULT_SOFT_CACHE_SIZE);
	}

	private Cache getCache(Key key) {
		int h = key.hashCode();
        h ^= (h >>> 21) ^ (h >>> 13);
        h ^= (h >>> 8) ^ (h >>> 5);

        h &= NUM_CACHES - 1;
        return caches[h];
	}
	
	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.storage.AbstractProbTreeStorage#getProbTree(int, long)
	 */
	@Override
	public OnDiskCompactProbTree getProbTree(Key key) {
		RequestResult result = getCache(key).getItem(key);
		return result.tree;
	}

	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.storage.AbstractProbTreeStorage#setProbTree(int, long, edu.umd.clip.lm.model.AbstractProbTree)
	 */
	@Override
	public void setProbTree(Key key, OnDiskCompactProbTree probTree) {
		throw new UnsupportedOperationException("Cached storage only supports reading");
	}

	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.storage.AbstractProbTreeStorage#getStats()
	 */
	@Override
	public String getStats() {
		StringBuilder sb = new StringBuilder("Cache Stats: ");
		long hits = 0;
		long requests = 0;
		long pending = 0;
		long delayed = 0;
		long totalLockWaitTime = 0;
		for(Cache cache : caches) {
			hits += cache.getHits();
			requests += cache.getRequests();
			pending += cache.getPending();
			delayed += cache.getDelayed();
			totalLockWaitTime += cache.getLockWaitTime();
		}
		NumberFormat nf = new DecimalFormat("#.##");
		
		sb.append(String.format("%s%% hits of %d, pending=%d, %d locks waited for %sms avg", 
				nf.format(hits * 100.0 /requests) , requests, pending, delayed, 
				nf.format(totalLockWaitTime * 1e-6 / delayed)));

		sb.append("\nBackend Stats:\n");
		sb.append(backend.getStats());
		return sb.toString();
	}

	@Override
	protected void requestProbTree(Key key) {
		if (!getCache(key).containsItem(key)) {
			backend.requestProbTree(key);
		}
	}

	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.storage.AsyncProducer#request(java.lang.Object, edu.umd.clip.lm.storage.AsyncConsumer)
	 */
	@Override
	public void request(Key key, AsyncConsumer<Key, OnDiskCompactProbTree> callback) {
		Cache cache = getCache(key);
		cache.request(key, callback);
	}

	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.storage.AbstractProbTreeStorage#closeAll()
	 */
	@Override
	public void closeAll() {
		backend.closeAll();
	}
	
}
